package com.atguigu.tingshu.search.task;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.BaseCategoryFeignClient;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory1;
import com.atguigu.tingshu.search.service.ItemService;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @Author GQS @Date 2024/11/6/0:46 @Description
 * 同步专辑数据到es中去消费消息失败的兜底处理
 */
@Component
public class AlbumIndexInfoMessageTask {

  @Autowired
  private RedisTemplate redisTemplate;

  @Resource
  private AlbumInfoFeignClient albumInfoFeignClient;

  @Autowired
  private ItemService itemService;

  /**
   * 消费同步消息失败的兜底方案
   */
  //  @Scheduled(cron = "0 0/30 * * * ?")
  @XxlJob("albumIndexInfoMessageTask")
  // @Scheduled(fixedDelay = 30000)
  public void albumIndexInfoMessageTask(){

    //从redis中获取所有失败的专辑的id信息
    Set<String> albumIdSet =
            redisTemplate.opsForSet().members("search_upper_or_down");
    //遍历处理
    albumIdSet.stream().forEach(albumId->{
      try {
        Long id = Long.valueOf(albumId);
        //查询专辑的数据
        AlbumInfo albumInfo = albumInfoFeignClient.getAlbumInfo(id);
        //判断专辑是否为公开
        if(albumInfo.getIsOpen().equals("1")){
          //公开,写入es
          itemService.albumInfoFromMysqlToEs(id);
        }else{
          //不公开,删除es中专辑的数据
          itemService.removeAlbumInfoFromEs(id);
        }
        //删除这个id
        redisTemplate.opsForSet().remove("search_upper_or_down", albumId);
      }catch (Exception e){
        e.printStackTrace();
      }
    });
  }
  /**
   * 定时任务: 每天的凌晨1点执行,同步前一天的全部变化的专辑的数据
   */
  @XxlJob("updateEsAlbumStat")
  public void updateEsAlbumStat(){
    //获取当前的时间
    String dateTime = new DateTime().plusDays(-1).toString("yyyyMMdd");
    //获取所有需要同步的专辑的id
    Set<String> members =
            redisTemplate.opsForSet().members("Album_Stat_Change_Set_" + dateTime);
    if(members !=null && members.size() > 0){
      //遍历统计
      members.stream().forEach(albumIdString->{
        //同步数据
        itemService.updateByTask(Long.valueOf(albumIdString));
      });
    }
  }


  @Autowired
  private BaseCategoryFeignClient baseCategoryFeignClient;

  @Autowired
  private KafkaTemplate kafkaTemplate;


  /**
   * 常量定义了专辑的用户四种行为
   */
  private static final String[] types=
          new String[]{"playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum", "hotScore"};

  /**
   * 排行榜数据每日同步定时任务
   */
  @XxlJob("dayRankTask")
  public void dayRankTask(){
    //最终结果集初始化
    Map<String, Object> result = new ConcurrentHashMap();
    //播放集合
    List<JSONObject> playList = new CopyOnWriteArrayList<>();
    //收藏集合
    List<JSONObject> collectList = new CopyOnWriteArrayList<>();
    //评论集合
    List<JSONObject> commentList = new CopyOnWriteArrayList<>();
    //购买集合
    List<JSONObject> buyList = new CopyOnWriteArrayList<>();
    //计算日-1的时间
    String time = new DateTime().plusDays(-1).toString("yyyyMMdd");
    //查询所有的一级分类
    List<BaseCategory1> baseCategory1List =
            baseCategoryFeignClient.getBaseCategory1();
    //维度遍历
    for (int i = 0; i < types.length; i++) {
      //获取行为
      String type = types[i];
      //遍历所有的一级分类
      baseCategory1List.stream().forEach(category1->{
        //拼接key
        String key = category1.getId() + ":" + type + ":" + time;
        //热度值就直接删
        if(!type.equals("hotScore")){
          //获取全部的长度
          Long size = redisTemplate.opsForZSet().size(key);
          //获取全部的数据
          Set<Long> albumIdSet = redisTemplate.opsForZSet().range(key, 0, size - 1);
          albumIdSet.stream().forEach(albumId -> {
            JSONObject json = new JSONObject();
            json.put("albumId", albumId);
            //获取每个专辑的每日的每个维度的分数
            int num = redisTemplate.opsForZSet().score(key, albumId).intValue() * -1;
            json.put("num", num);
            //新增到list
            switch (type){
              case "playStatNum"->playList.add(json);
              case "subscribeStatNum"->collectList.add(json);
              case "buyStatNum"->buyList.add(json);
              case "commentStatNum"->commentList.add(json);
            }
          });
        }
        redisTemplate.delete(key);
      });
    }
    //整合结果集: 日表设计表结构: id 类型 主体id 数量 日期 创建时间 修改时间 逻辑字段
    kafkaTemplate.send("album_stat_topic", 0, "play", JSONObject.toJSONString(playList));
    kafkaTemplate.send("album_stat_topic", 1, "collect", JSONObject.toJSONString(collectList));
    kafkaTemplate.send("album_stat_topic", 2, "buy", JSONObject.toJSONString(buyList));
    kafkaTemplate.send("album_stat_topic", 3, "comment", JSONObject.toJSONString(commentList));
  }
}
